// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#include <unordered_map>

#include <cxxopts.hpp>
#include <boost/asio.hpp>
#include <boost/asio/high_resolution_timer.hpp>

#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/executor/topic_distributor.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/market/enums.h"
#include "coin2/strategy/strategy.h"
#include "coin2/flow/writer.h"
#include "coin/proto/coin_market_query.pb.h"

using namespace coin2::exchange::base::symbology;
using namespace coin2::base::config;
using namespace coin2::exchange::feed;
using namespace coin2::app;
using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::base::market::MarketExchangeApi;
using coin2::exchange::base::market::ExchangeTypeToString;
using coin2::exchange::base::market::MarketTypeToString;
using coin2::exchange::base::executor::TopicRecordProcessor;

namespace {
struct ProductTicker {
  std::string symbol;
  double mid_price;
  int64_t timestamp;
};

struct ExchangeTicker {
  MarketExchangeApi mea;
  std::unordered_map<std::string, ProductTicker> product_tickers;
};
}

class GenerateTickerStrategy : public ::coin2::strategy::IStrategy {
 public:
  GenerateTickerStrategy(
      const std::string& out_dir,
      const std::string& machine)
      : out_dir_(out_dir),
        machine_(machine) {
  }

  void Init(::coin2::app::Driver* driver) {
    generate_ticker_timer_ = std::make_unique<boost::asio::deadline_timer>(
        *driver->live_executor()->MainIoc());
    generate_ticker_timer_->expires_from_now(boost::posix_time::nanoseconds(
        30 * 1'000'000'000LL));
    generate_ticker_timer_->async_wait(std::bind(
        &GenerateTickerStrategy::GenerateTicker, this));
  }

  void onBookFeed(const FeedUpdate& upd) override {
    auto& book = upd.book();
    if (book.Ask0() && book.Bid0()) {
      const auto& mea = upd.mea();
      if (exchange_tickers_.count(mea) == 0) {
        exchange_tickers_.insert({mea, ExchangeTicker{
            mea,
            std::unordered_map<std::string, ProductTicker>()
        }});
      }
      auto& exchange_ticker = exchange_tickers_.at(mea);
      const auto& symbol = upd.product().symbol();
      if (exchange_ticker.product_tickers.count(symbol) == 0) {
        exchange_ticker.product_tickers.insert({symbol, ProductTicker{
            symbol,
            0,
            0
        }});
      }
      auto& product_ticker = exchange_ticker.product_tickers.at(symbol);
      product_ticker.mid_price = (book.Ask0()->price + book.Bid0()->price) / 2;
      product_ticker.timestamp = upd.timestamp();
    }
  }

  void GenerateTicker() {
    for (const auto& [mea, exchange_ticker] : exchange_tickers_) {
      int64_t timestamp = TopicRecordProcessor::GetTimestamp();
      coin::proto::SubscriptionLog proto;
      proto.set_type(coin::proto::SubscriptionLog::TICKER);
      proto.mutable_request()->set_market_type(MarketTypeToString(exchange_ticker.mea.market));
      proto.mutable_request()->set_exchange(ExchangeTypeToString(exchange_ticker.mea.exchange));
      proto.mutable_request()->set_api_version(exchange_ticker.mea.api);
      proto.mutable_ticker()->set_exchange(ExchangeTypeToString(exchange_ticker.mea.exchange));
      proto.mutable_ticker()->set_market_type(MarketTypeToString(exchange_ticker.mea.market));
      proto.mutable_ticker()->set_api_version(exchange_ticker.mea.api);
      proto.mutable_ticker()->set_query_ts(timestamp);
      for (const auto& [product, ticker] : exchange_ticker.product_tickers) {
        auto product_ticker = proto.mutable_ticker()->add_each_ticker();
        product_ticker->set_symbol(ticker.symbol);
        product_ticker->set_last(ticker.mid_price);
        product_ticker->set_exchange_timestamp(ticker.timestamp);
      }

      std::string buf;
      buf.resize(orio::record::TopicRecord::struct_size_with_data(proto.ByteSize()));
      auto* r = reinterpret_cast<orio::record::TopicRecord*>(buf.data());
      r->data_len = proto.ByteSize();
      r->queue_seq = 0;
      r->topic_id = 0;
      r->topic_seq = 0;
      r->timestamp = timestamp;

      std::string buf2;
      proto.SerializeToString(&buf2);
      const auto* data = reinterpret_cast<const std::byte*>(buf2.data());
      std::copy(data, data + r->data_len, r->data);

      if (writers_.find(mea) == std::end(writers_)) {
        writers_[mea] = std::make_unique<FlowWriter>(
            out_dir_, machine_, mea.SnakeString());
      }

      FlowWriter* writer = writers_[mea].get();
      writer->Write(*r);
      writer->Flush();
    }
    generate_ticker_timer_->expires_from_now(boost::posix_time::nanoseconds(
        30 * 1'000'000'000LL));  // 30 sec
    generate_ticker_timer_->async_wait(std::bind(
        &GenerateTickerStrategy::GenerateTicker, this));
  }

 private:
  std::string out_dir_;
  std::string machine_;
  std::unordered_map<MarketExchangeApi, std::unique_ptr<FlowWriter>> writers_;
  std::unique_ptr<boost::asio::deadline_timer> generate_ticker_timer_;
  std::unordered_map<MarketExchangeApi, ExchangeTicker> exchange_tickers_;
};

int main(int argc, char* argv[]) {
  google::InitGoogleLogging(argv[0]);

  DriverConfig app;

  cxxopts::Options opt("Ticker publisher", "Generate ticker proto feed.");
  opt.add_options()(
      "out_dir",
      "ticker out directory",
      cxxopts::value<std::string>()->default_value("/data/feed-data/ticker/"));
  opt.add_options()(
      "machine",
      "machine",
      cxxopts::value<std::string>()->default_value(""));
  AddDriverOptions(&opt);
  AddIntervalOptions(&opt);
  AddProductOptions(&opt);

  auto res = opt.parse(argc, argv);
  ParseDriverOptions(res, &app);
  ParseIntervalOptions(
      res,
      app.mutable_feed()->mutable_common()->mutable_archive()->mutable_interval());
  ParseProductOptions(res, app.mutable_feed());

  LOG(INFO) << app.DebugString();

  CHECK(!res["machine"].as<std::string>().empty());
  GenerateTickerStrategy ticker_generator(
      res["out_dir"].as<std::string>(),
      res["machine"].as<std::string>());
  Driver driver(app, &ticker_generator);
  driver.Run();

  return 0;
}
